[cdc] Update flink.cdc.version to 3.5.0 and add FlinkCDC to Paimon converter. by lvyanquan · Pull Request #6358 · apache/paimon

lvyanquan · 2025-09-30T09:15:56Z

Purpose

Linked issue: close #6357

Tests

API and Format

Documentation

yunfengzhou-hub

Thanks for the PR. Left some comments as below.

yunfengzhou-hub · 2025-11-04T02:46:12Z

...c/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCDataConverter.java

+            case INSERT:
+            case UPDATE_AFTER:
+                {
+                    return DataChangeEvent.insertEvent(tableId, binaryRecordData, new HashMap<>());


Should it be DataChangeEvent.updateEvent? Same for the UPDATE_BEFORE below.

Flink Row encodes an update event as two separated rows (-U and +U), while CDC UpdateEvent carries both as one event. Seems it's not possible to build a complete DataChangeEvent.updateEvent with given UPDATE_BEFORE or UPDATE_AFTER row here.

yunfengzhou-hub · 2025-11-04T02:48:27Z

...c/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCDataConverter.java

+                    return DataChangeEvent.deleteEvent(tableId, binaryRecordData, new HashMap<>());
+                }
+            default:
+                throw new IllegalArgumentException("don't support type of " + row.getRowKind());


nit: Capitalize the first letter of don't. Same as FlinkCDCToPaimonDataConverter.

yunfengzhou-hub · 2025-11-04T02:51:05Z

...c/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCDataConverter.java

+                fieldGetter = row -> row.getInt(fieldPos);
+                break;
+            case DATE:
+                fieldGetter = row -> row.getInt(fieldPos);


Should the result type be something like DateTime or Instant? Same for below

yunfengzhou-hub · 2025-11-04T02:54:02Z

...c/src/main/java/org/apache/paimon/flink/pipeline/cdc/util/PaimonToFlinkCDCDataConverter.java

+                        "don't support type of " + fieldType.getTypeRoot());
+        }
+        if (!fieldType.isNullable()) {
+            return fieldGetter;


Let's add a test to verify the behavior when values are null but types require not null.

Some quick experiments show that a "default" value will be filled when trying to write a null value into NOT NULL fields, like false for BOOLEAN and 1970-01-01 for DATE.

Shall we a) keep this behavior and add test to cover this behavior, or b) validate nullability violations and throw an exception if occurs?

yunfengzhou-hub · 2025-11-04T02:55:59Z

...aimon-flink-cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/util/DataConvertTest.java

+        org.apache.flink.cdc.common.event.DataChangeEvent dataChangeEvent =
+                DataChangeEvent.insertEvent(tableId, recordDataGenerator.generate(testData));
+
+        Assertions.assertEquals(


Let's also verify the converted data in the middle phase, after the original CDC data is converted to paimon data and before it's converted back.

yunfengzhou-hub · 2025-11-04T03:00:00Z

...ink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/PaimonMetadataApplier.java

+            } else if (schema.partitionKeys() != null && !schema.partitionKeys().isEmpty()) {
+                partitionKeys.addAll(schema.partitionKeys());
+            }
+            builder.primaryKey(primaryKeys)


I noticed that in Flink CDC, this implementation originally looks like this

for (String partitionColumn : partitionKeys) { if (!primaryKeys.contains(partitionColumn)) { primaryKeys.add(partitionColumn); } } builder.partitionKeys(partitionKeys) .primaryKey(primaryKeys)

Why do we change the implementation here?

Added it back for consistency.

yunfengzhou-hub · 2025-11-04T03:17:11Z

...cdc/src/test/java/org/apache/paimon/flink/pipeline/cdc/schema/PaimonMetadataApplierTest.java

+    }
+
+    @Test
+    public void testMysqlDefaultTimestampValueConversionInAddColumn()


Is "Mysql" a typo? Or maybe better to add a comment describing why this paimon test is related to mysql.

I think the original test case is meant to cover some special default values in MySQL timestamp fields (like CURRENT_TIMESTAMP). Renamed it to avoid confusion.

yunfengzhou-hub · 2025-11-04T03:18:42Z

...rc/test/java/org/apache/paimon/flink/action/cdc/mongodb/MongoDBSyncDatabaseActionITCase.java


        waitingTables("t3");
-        jobClient.cancel();
+        jobClient.cancel().get();


Better add a timeout to avoid infinite blocking. Same for other test cases.

yunfengzhou-hub · 2025-11-04T03:20:37Z

...link-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/schema/SchemaChangeProvider.java

+     * @param value The value of the option to be set.
+     * @return A SchemaChange object representing the setting of an option.
+     */
+    public static SchemaChange setOption(String key, String value) {


unused method.

yunfengzhou-hub · 2025-11-04T03:23:57Z

...flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/postgres/PostgresActionUtils.java

        CdcDebeziumDeserializationSchema schema =
                new CdcDebeziumDeserializationSchema(true, customConverterConfigs);
-        return sourceBuilder.deserializer(schema).includeSchemaChanges(true).build();
+        return sourceBuilder.deserializer(schema).build();


Shall we describe why we need to make changes like this in non-Paimon-Flink-CDC classes? I suppose these changes are along with the Flink CDC version upgrade from 3.1 to 3.5, but not quite sure why we need to make such changes. We can add some details to the description section of this PR.

IIUC due to technical limitations, PostgreSQL CDC does not support capturing DDL events or emitting any schema change events. In earlier versions this method (includeSchemaChanges) is presented but has no effect at all, and got removed in apache/flink-cdc#3464.

yuxiqian · 2025-11-05T01:30:14Z

Thanks @yunfengzhou-hub for kindly reviewing this. As Yanquan is on vacation right now, I'll draft another PR and address comments there.

JingsongLi · 2025-12-28T04:04:20Z

Implemented in #6532

lvyanquan and others added 5 commits November 3, 2025 16:13

[cdc] Update flink.cdc.version to 3.5.0.

0e68103

[cdc] Add PaimonMetadataApplier.

aee914c

[cdc] Add TypeConverter and DataConverter.

2a0c861

[cdc] Add TypeConverter and DataConverter.

9838600

Fix unstable CI

529b740

yunfengzhou-hub mentioned this pull request Nov 3, 2025

[Feature] Update Flink dependency to 2.0 #5350

Open

2 tasks

lvyanquan added 4 commits November 3, 2025 22:40

[cdc] Update flink.cdc.version to 3.5.0.

065d61a

[cdc] Add PaimonMetadataApplier.

bc544e5

[cdc] Add TypeConverter and DataConverter.

5b4eaab

[cdc] Add TypeConverter and DataConverter.

1176869

lvyanquan force-pushed the cdc-version branch from 8b8b2fa to 1176869 Compare November 3, 2025 14:41

[cdc][contrib] Fixes CDC test case failure

bb323b3

yunfengzhou-hub reviewed Nov 4, 2025

View reviewed changes

yuxiqian mentioned this pull request Nov 5, 2025

[cdc] Bump Flink CDC version & migrate converter utils to Paimon repo #6532

Merged

JingsongLi closed this Dec 28, 2025

Conversation

lvyanquan commented Sep 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Tests

API and Format

Documentation

Uh oh!

yunfengzhou-hub left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yuxiqian Nov 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yuxiqian commented Nov 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

JingsongLi commented Dec 28, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

lvyanquan commented Sep 30, 2025 •

edited

Loading

yuxiqian Nov 5, 2025 •

edited

Loading

yuxiqian commented Nov 5, 2025 •

edited

Loading